package com.iacrqq.sedaf.stage;

import java.util.LinkedList;
import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.iacrqq.sedaf.config.Configuration;
import com.iacrqq.sedaf.event.Event;
import com.iacrqq.sedaf.event.EventDispatcher;
import com.iacrqq.sedaf.event.EventHandler;
import com.iacrqq.sedaf.event.exception.EventHandlerException;
import com.iacrqq.sedaf.executor.DefaultThreadPoolExecutor;
import com.iacrqq.sedaf.executor.Executor;
import com.iacrqq.sedaf.queue.Input;
import com.iacrqq.sedaf.queue.Output;
import com.iacrqq.sedaf.queue.exception.QueueFullException;

/**
 * 
 * @author sihai
 * 
 */
public class DefaultStage implements Stage
{

	private static final Log logger = LogFactory.getLog(DefaultStage.class);

	private String name;
	private EventHandler eventHandler;
	private EventDispatcher eventDispatcher;
	private Configuration configuration;
	private Input<Event> eventInput;
	private List<Output<Event>> outputs;
	private Output<Event> errorOutput;
	private Executor executor;

	private StageRuntimeStatisticsCollector runtimeStatisticsCollector;

	public DefaultStage(String name, EventHandler eventHandler, Input<Event> eventInput, List<Output<Event>> outputs,
			Output<Event> errorOutput)
	{
		this.outputs = new LinkedList<Output<Event>>();
		runtimeStatisticsCollector = new StageRuntimeStatisticsCollector(
				StageRuntimeStatisticsCollector.DEFAULT_HISTORY_COUNT);
		this.name = name;
		this.eventHandler = eventHandler;
		this.eventInput = eventInput;
		this.outputs.addAll(outputs);
		this.errorOutput = errorOutput;
		executor = new DefaultThreadPoolExecutor(new StringBuilder(name).append(".executor").toString(), this);
	}

	@Override
	public void init(Configuration config) throws Exception
	{
		this.configuration = config;
		eventHandler.init(config);
		executor.init(config);
	}

	@Override
	public Input<Event> getInput()
	{
		return this.eventInput;
	}

	@Override
	public void addOutput(Output<Event> output)
	{
		outputs.add(output);
	}
	
	@Override
	public List<Output<Event>> getOutputs() 
	{
		return outputs;
	}

	@Override
	public EventHandler getEventHandler()
	{
		return eventHandler;
	}

	@Override
	public void setEventHandler(EventHandler eventHandler)
	{
		this.eventHandler = eventHandler;
	}
	
	@Override
	public void setEventDispatcher(EventDispatcher eventDispatcher)
	{
		this.eventDispatcher = eventDispatcher;
	}

	@Override
	public void setOutputs(List<Output<Event>> outputs)
	{
		this.outputs = outputs;
	}

	@Override
	public String getName()
	{
		return name;
	}

	@Override
	public void setName(String name)
	{
		this.name = name;
	}

	@Override
	public int getCurrentThreadCount()
	{
		return executor.getCurrentThreadCount();
	}

	@Override
	public int getIdleTreadCount()
	{
		return executor.getIdleTreadCount();
	}

	@Override
	public StageRuntimeStatistics getStageRuntimeStatistics()
	{
		return runtimeStatisticsCollector.snapshot(this);
	}

	@Override
	public void destroy() throws Exception
	{
		executor.destroy();
		eventHandler.destroy();
		outputs.clear();
	}

	@Override
	public void pause()
	{
		executor.pause();
	}

	@Override
	public void start()
	{
		executor.start();
	}

	@Override
	public void stop()
	{
		executor.stop();
		logger.debug(runtimeStatisticsCollector.toString());
	}
	
	@Override
	public int idleThread(int num)
	{
		return this.executor.idleThread(num);
	}

	@Override
	public int addThread(int num)
	{
		return this.executor.addThread(num);
	}

	@Override
	public int removeThread(int num)
	{
		return this.executor.removeThread(num);
	}

	@Override
	public void run()
	{
		runtimeStatisticsCollector.start();
		while (!Thread.currentThread().isInterrupted())
		{
			Event event = null;

			if (eventInput != null)
			{
				event = eventInput.dequeue();
			}

			try
			{
				if (event != null || (event == null && eventInput == null))
				{
					// logger.debug(new
					// StringBuilder("Thread:").append(Thread.currentThread().getName()).append(" handle one event:").append(event).toString());
					runtimeStatisticsCollector.startHandle();
					Event newEvent = this.eventHandler.handleEvent(event, eventDispatcher);
					runtimeStatisticsCollector.endHandle();
					if (newEvent != null)
					{
						for (Output<Event> output : outputs)
						{
							while (true)
							{
								try
								{
									output.enqueue(newEvent);
									break;
								}
								catch (QueueFullException e)
								{
									runtimeStatisticsCollector.full();
									logger.error(e);
									// try again
									try
									{
										Thread.sleep(1000);
									}
									catch (InterruptedException ex)
									{
										logger.error(ex);
										Thread.currentThread().interrupt();
									}
									continue;
								}
							}
						}
					}

					runtimeStatisticsCollector.completed();
				}
				else
				{
					runtimeStatisticsCollector.nop();
					//
					// Thread.currentThread().yield(); // not always give up cpu
					try
					{
						Thread.sleep(100);
					}
					catch (InterruptedException ex)
					{
						logger.error(ex);
						Thread.currentThread().interrupt();
					}
					continue;
				}
			}
			catch (EventHandlerException e)
			{
				runtimeStatisticsCollector.endHandle();
				logger.error(e);
				try
				{
					errorOutput.enqueue(event);
				}
				catch (QueueFullException ex)
				{
					logger.error(ex);
					// try again
					try
					{
						Thread.sleep(1000);
					}
					catch (InterruptedException exx)
					{
						logger.error(exx);
						Thread.currentThread().interrupt();
					}
					continue;
				}
			}
		}

		runtimeStatisticsCollector.end();
	}
}
